#include "outgoing_connection.h"
#include "util.h"
#include "constants.h"

using namespace std;

OutgoingConnection::OutgoingConnection(Server *server, string address,
                                       int socket, pollfd *poll_info) :
	Connection(socket, poll_info), state(SENDING_MESSAGE),
	server(server), address(address), file(0), reply_state(AWAITING_ERROR_CODE) {
	filter_events(POLLOUT);
	log("Connecting to client to retrieve file information.");
	server->add_connection_to(address, this);
}

static const int BUFSIZE = 1024;

void OutgoingConnection::process_events() {
	static char buf[BUFSIZE];
	if (state == SENDING_MESSAGE && can_send()) {
		ssize_t numbytes = send(command.c_str(), command.length());
		if (numbytes < 0) {
			error("Error writing to socket");
		} else {
			command.erase(0, numbytes);
			if (command.empty()) {
				change_state(AWAITING_REPLY);
				reply_state = AWAITING_ERROR_CODE;
			}
		}
	}
	if (state == AWAITING_REPLY && can_receive()) {
		ssize_t numbytes = receive(buf, BUFSIZE);
		if (numbytes < 0) {
			error("Error reading from socket");
		} else if (numbytes == 0) {
			log("Connection closed by client.");
			kill_connection();
		} else {
			buffer += string(buf, numbytes);
			size_t pos;
			while ((pos = buffer.find(separator)) != string::npos) {
				process_reply(buffer.substr(0, pos));
				buffer.erase(0, pos + separator.length());
			}
		}
	}
}

void OutgoingConnection::enqueue_file(const std::string md5,
                                      const Source *source) {
	for (size_t i = 0; i < queue.size(); i++) {
		if (queue[i].first == md5) {
			vector<const Source *> &v = queue[i].second;
			if (find(v.begin(), v.end(), source) == v.end()) {
				v.push_back(source);
			}
			return;
		}
	}
	queue.push_back(make_pair(md5, vector<const Source *>(1, source)));
	if (queue.size() == 1) {
		log("Requestin data for file " + md5);
		command = "get_metadata " + md5 + separator;
	}
}


void OutgoingConnection::change_state(FSMState newstate) {
	state = newstate;
	if (newstate == SENDING_MESSAGE) filter_events(POLLOUT);
	if (newstate == AWAITING_REPLY) filter_events(POLLIN);
}

void OutgoingConnection::process_reply(const string &reply) {
	if (reply_state == AWAITING_ERROR_CODE) {
		if (is_reply_successful(reply)) {
			reply_state = AWAITING_NAME;
		} else {
			process_next_file();
		}
	} else if (reply_state == AWAITING_NAME) {
		file = new File(queue.front().first, reply);
		reply_state = AWAITING_SIZE;
	} else if (reply_state == AWAITING_SIZE) {
		file->set_file_size(string_to_int(reply));
		reply_state = AWAITING_FRAGMENT_MD5;
	} else if (reply_state == AWAITING_FRAGMENT_MD5) {
		if (!reply.empty()) {
			Fragment *fragment = new Fragment(reply);
			for (size_t i = 0; i < queue.front().second.size(); i++) {
				fragment->add_source(queue.front().second[i]);
			}
			file->add_fragment(fragment);
		} else {
			// Agregarlo al indice.
			if (server && static_cast<int>(file->get_fragments().size()) ==
			    (file->get_size() + BLOCK_SIZE - 1) / BLOCK_SIZE) {
				server->get_index().add_file(file);
			} else {
				delete file;
			}
			process_next_file();
		}
	}
}

void OutgoingConnection::process_next_file() {
	queue.pop_front();
	if (queue.empty()) {
		kill_connection();
	} else {
		log("Requesting data for file " + queue.front().first);
		command = "get_metadata " + queue.front().first + separator;
		change_state(SENDING_MESSAGE);
	}
}

OutgoingConnection::~OutgoingConnection() {
	if (server) {
		server->remove_connection_to(address);
	}
}
